/*
 * Copyright (c) 2017, www.zm0618.com All Rights Reserved..
 */

package com.ibetter;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.Iterator;

/**
 * Created by zhaojun on 2017/03/01.
 */
public class App2 {

    public static void main(String[] args) throws IOException {
        Job job = Job.getInstance();

        job.setJobName("app2");

        job.setInputFormatClass(TextInputFormat.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        FileInputFormat.addInputPath(job,new Path("file:///F:/mr/input"));
        FileOutputFormat.setOutputPath(job,new Path("file:///F:/mr/out1"));

        try {
            job.waitForCompletion(true);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    static class MyMapper extends Mapper<LongWritable,Text,Text,LongWritable>{

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String lineVal = value.toString();
            if (StringUtils.isNotBlank(lineVal)){
                String[] vals = lineVal.split("=");
                context.write(new Text(vals[0]),new LongWritable(Long.parseLong(vals[1])));
            }
        }
    }

    static class MyReducer extends Reducer<Text,LongWritable,Text,LongWritable>{

        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

            Iterator<LongWritable> ite = values.iterator();
            long count=0;
            while (ite.hasNext()){
                count+=ite.next().get();
            }
            context.write(key,new LongWritable(count));
        }
    }
}
